package com.lfz.search.rabbit;

import com.lfz.common.service.RedisService;
import com.lfz.search.service.JobItemService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import static com.lfz.search.config.Constant.JOB_ITEM_PUT_QUEUE_NAME;
import static com.lfz.search.config.Constant.JOB_ITEM_SEARCH_QUEUE_NAME;

/**
 * @author Daniel
 */
@Component
public class RabbitServer {

    @Autowired
    private JobItemService jobItemService;

    @Autowired
    private RedisService redisService;

    @Autowired
    AmqpTemplate amqpTemplate;

    @Autowired
    AsyncTask asyncTask;

    Logger logger = LoggerFactory.getLogger(getClass());

    @RabbitListener(queues = JOB_ITEM_PUT_QUEUE_NAME)
    public void processJobItemPut(String message, @Header(AmqpHeaders.REPLY_TO) String replyTo) {
        // logger.info("recevie message {} and reply to {}", message, replyTo);
        // if(replyTo.startsWith("amq.rabbitmq.reply-to")) {
        //     logger.debug("starting with version 3.4.0, the RabbitMQ server now supports Direct reply-to");
        // }else {
        //     logger.info("fall back to using a temporary reply direct");
        // }
        ListenableFuture<String> asyncResult = asyncTask.PutJobItemOperation(message);
        asyncResult.addCallback(new ListenableFutureCallback<String>() {
            @Override
            public void onSuccess(String result) {
                amqpTemplate.convertAndSend(replyTo, result);
            }

            @Override
            public void onFailure(Throwable ex) {

            }
        });
    }

    @RabbitListener(queues = JOB_ITEM_SEARCH_QUEUE_NAME)
    public void processJobItemSearch(String message, @Header(AmqpHeaders.REPLY_TO) String replyTo) {
        // logger.info("recevie message {} and reply to {}", message, replyTo);
        // if(replyTo.startsWith("amq.rabbitmq.reply-to")) {
        //     logger.debug("starting with version 3.4.0, the RabbitMQ server now supports Direct reply-to");
        // }else {
        //     logger.info("fall back to using a temporary reply direct");
        // }
        ListenableFuture<String> asyncResult = asyncTask.searchJobItemOperation(message);
        asyncResult.addCallback(new ListenableFutureCallback<String>() {
            @Override
            public void onSuccess(String result) {
                amqpTemplate.convertAndSend(replyTo, result);
            }

            @Override
            public void onFailure(Throwable ex) {

            }
        });
    }


}
